package org.lds.message.center.admin.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.ProducerListener;

@Slf4j
@Configuration
public class KafkaConfiguration {

    @Bean
    public KafkaProducer kafkaProducer(ProducerFactory<String, String> producerFactory) {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory);
        kafkaTemplate.setProducerListener(new KafkaProducerListener());
        return new KafkaProducer(kafkaTemplate);
    }

    private class KafkaProducerListener implements ProducerListener<String, String> {
        @Override
        public void onError(ProducerRecord<String, String> producerRecord, RecordMetadata recordMetadata, Exception exception) {
            log.info("数据发送失败 message={} cause={}", producerRecord.value(), exception.getMessage());
        }
    }

}
